package com.ry.flink.job5;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;
/**
 * 需求：单词每出现三次统计一次
 */
public class GlobalWindowTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> dataStream = env.socketTextStream("192.168.134.130", 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> stream = dataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] fields = line.split(",");
                for (String word : fields) {
                    collector.collect(Tuple2.of(word, 1));
                }
            }
        });

        stream.keyBy(0)
                //自定的CountTrigger跟源码非常相识
//                .countWindow(2,3)
                //如果不配合 trigger 不会运行该窗口
                .window(GlobalWindows.create())
                .trigger(new CountTrigger(3))
                .sum(1)
                .print();

        env.execute("SessionWindowTest");
    }

    /**
     *自定义一个触发策略: 数据的个数达到某个值的时候，触发计算
     * state：用来统计数据的个数
     */
    private static class CountTrigger extends Trigger<Tuple2<String,Integer>, GlobalWindow>{

        private long maxCount;

        // 注册状态
        ReducingStateDescriptor<Long> descriptor =
                new ReducingStateDescriptor<Long>(
                        "count",  // 状态的名字
                        new ReduceFunction<Long>() { // 聚合函数
                            @Override
                            public Long reduce(Long value1, Long value2) throws Exception {
                                return value1 + value2;
                            }
                        }, Long.class); // 状态存储的数据类型


        public CountTrigger(long maxCount){
            this.maxCount = maxCount;
        }

        /**
         *
         * 进入窗口的每个元素都会调用该方法
         * @param element
         * @param timestamp
         * @param window
         * @param ctx
        CONTINUE 表示对窗口不进行任何处理
        FIRE_AND_PURGE ： 触发window的计算，并且清除当前window的数据
        FIRE 触发window的计算
        PURGE 清除window中所有的数据

         *
         */
        @Override
        public TriggerResult onElement(Tuple2<String, Integer> element,
                                       long timestamp, GlobalWindow window,
                                       TriggerContext ctx) throws Exception {

            //获取当前key对应的count
            ReducingState<Long> count = ctx.getPartitionedState(descriptor);
            count.add(1L);
            if(count.get() == maxCount){
                count.clear();
                return TriggerResult.FIRE_AND_PURGE;
            }

            return TriggerResult.CONTINUE;
        }

        /**
         * 事件时间timer触发的时候被调用
         * */
        @Override
        public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }

        /**
         * 处理时间timer触发的时候会被调用。
         * */
        @Override
        public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }

        /**
         * 执行窗口及状态数据的清除方法
         * */
        @Override
        public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {

            ctx.getPartitionedState(descriptor).clear();
        }
    }
}
